package com.atguigu.userprofile.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.userprofile.bean.TagCondition;
import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.UserGroup;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.mapper.UserGroupMapper;
import com.atguigu.userprofile.service.TagInfoService;
import com.atguigu.userprofile.service.UserGroupService;
import com.atguigu.userprofile.utils.RedisUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;

import org.apache.catalina.User;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2021-05-04
 */
@Service
@Slf4j
@DS("mysql")
public class UserGroupServiceImpl extends ServiceImpl<UserGroupMapper, UserGroup> implements UserGroupService {

    @Autowired
    private TagInfoService tagInfoService;
    /*
        生成求多个分群交集的sql

            select  bitmapAnd(
                       bitmapAnd(
                        (sql1), (sql2)
                        ),
                        (sql3)
                   )

             bitmapAnd: 两两求交易
                bitmapAnd 的数量 = TagCondition的数量 - 1
     */
    public String generateBitmapAndSql(UserGroup userGroup){

        //获取所有的查询条件
        List<TagCondition> tagConditions = userGroup.getTagConditions();
        //查询所有标签的元数据Map
        Map<String, TagInfo> map = tagInfoService.getTagInfoMapWithCode();
        //获取业务日期
        String doDate = userGroup.getBusiDate();

        //准备拼接字符串的工具
        StringBuilder sqlBuilder = new StringBuilder();

        //生成语句
        for (TagCondition tagCondition : tagConditions) {

            String sql = generateSingleTagConditionSql(tagCondition, map, doDate);
            //当只有一个 tagCondition ，无需拼接 bitmapAnd
            if (sqlBuilder.length() == 0){
                sqlBuilder.append("(").append(sql).append(")");
            }else{
                sqlBuilder.insert(0," bitmapAnd(").append(",(").append(sql).append("))");
            }

        }

        String sql = sqlBuilder.toString();
        System.out.println(sql);
        return sql;

    }

    /*
        基于一个TagCondtion生成sql
        select
           groupBitmapMergeState(us)
         from user_tag_value_string     --表名是变量。 根据tagCondition查询的 tagCode的 tagValueType确定表名
                                                把目前定义的所有的Tag信息，全部缓存到一个Map集合中。
                                                    只需要从Map集合中根据tagCode获取TagInfo信息。

                                                  Map<String,TagInfo> map =  TagInfoServiceImpl.getTagInfoMapWithCode()


         where dt = '2020-06-14'        --业务日期由条件传入
                and
               tag_code='TAG_POPULATION_ATTRIBUTES_NATURE_GENDER'  -- tagCode
               and
              tag_value
                    in ('男性','女性')   -- 操作符由条件传入。操作符是in 或 notin 需要在值的外层拼接()
                                        -- 值由tagValues获取，如果是多个值，两两拼接 ,
                                        --  值是string 或 date类型，还需要拼接 '

     */
    public String generateSingleTagConditionSql(TagCondition tagCondition,Map<String,TagInfo> map,String doDate){

        String template = "select  groupBitmapMergeState(us)  from %s where dt ='%s'  and  tag_code='%s' and tag_value %s %s ";

        //需要查询到tag的 tagValueType
        TagInfo tagInfo = map.get(tagCondition.getTagCode());

        //确定从哪张表查询
        String tableName ="";

        String tagValueType = tagInfo.getTagValueType();

        switch (tagValueType){
            case ConstCodes.TAG_VALUE_TYPE_LONG: tableName = "user_tag_value_long" ; break;
            case ConstCodes.TAG_VALUE_TYPE_DECIMAL: tableName = "user_tag_value_decimal" ; break;
            case ConstCodes.TAG_VALUE_TYPE_STRING: tableName = "user_tag_value_string" ; break;
            case ConstCodes.TAG_VALUE_TYPE_DATE: tableName = "user_tag_value_date" ; break;
        }

        //确定操作符  operator:  eq,gt,in  ，在sql中要处理下， eq 翻译为 = ，gt 翻译为 >
        String sqlOperator = getConditionOperator(tagCondition.getOperator());

        String tagValueStr = "";
        //确定tagValueList
        if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagValueType) || ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagValueType)){
            //每个元素拼接 ' ，再两两拼接 ,    '00后','90后','60后'
            tagValueStr = "'" + StringUtils.join(tagCondition.getTagValues(),"','") +"'";
        }else{
            tagValueStr = StringUtils.join(tagCondition.getTagValues(),",") ;
        }

        //判断是否是范围查询
        if ("in".equals(tagCondition.getOperator()) || "nin".equals(tagCondition.getOperator())){
            tagValueStr = "(" + tagValueStr + ")" ;
        }

        //格式化
        String sql = String.format(template, tableName, doDate, tagCondition.getTagCode(), sqlOperator, tagValueStr);

        return sql;

    }

    private String getConditionOperator(String operator){

        switch (operator) {
            case "eq" : return "=";
            case "neq" : return "<>";
            case "gte" : return ">=";
            case "lte" : return "<=";
            case "gt" : return ">";
            case "lt" : return "<";
            case "in" : return "in";
            case "nin" : return "not in";
        }
        throw new RuntimeException("没有匹配的操作符!");

    }

    @Override
    public Long calulateUserGroupNum(UserGroup userGroup) {

        String sql = generateBitmapAndSql(userGroup);

        //将计算的分群结果写入ck
        super.baseMapper.insertUserGroupToCK(userGroup.getId(),sql);

        //查询刚刚写入的结果
        Long userGroupNum = super.baseMapper.queryUserGroupNum(userGroup.getId().toString());

        return userGroupNum;
    }

    @Override
    public void writeUserGroupToMysql(UserGroup userGroup) {

        Date createTime = new Date();
        userGroup.setCreateTime(createTime);

        //描述分群
        userGroup.setConditionComment(userGroup.conditionJsonToComment());

        //生成tagConditions 的json
        userGroup.setConditionJsonStr(JSON.toJSONString(userGroup.getTagConditions()));

        //写入mysql
        save(userGroup);

    }

    /*
            从ck中查询分群包

            写入redis
                    K:  唯一标识
                            id
                    V:  单值
                        集合: list,set,zset
                                set集合提供api: sismember ，方便地判断一个用户是否在一个分群中！
     */
    @Override
    public Long writeDataToRedis(UserGroup userGroup) {

        // 从ck中查询分群包   无需担心删除尚未执行
        List<String> users = super.baseMapper.queryUserGroup(userGroup.getId().toString());

        //获取连接
        Jedis jedis = RedisUtil.getJedis();

        int size = users.size();

        //把List转为Array
        String[] value= new String[size];

        for (int i = 0; i < value.length; i++) {

            value[i] = users.get(i);

        }

        // String... members: 称为可变参数，实际就是 String []
        String key = "user_grop:" + userGroup.getId();

        //先清空
        jedis.del(key);

        //向当前的key中追加
        jedis.sadd(key,value);

        return value.length + 0l;

    }

    @Override
    public Long evaluateUserGroupNum(UserGroup userGroup) {

        String sql = generateBitmapAndSql(userGroup);

        Long num = super.baseMapper.evaluateUserGroupNum(sql);

        return num;
    }

    @Override
    public void updateClickhouse(UserGroup userGroup) {

        Long id = userGroup.getId();
        //删除旧的人群包
        super.baseMapper.deleteUserGroup(id.toString());

        String sql = generateBitmapAndSql(userGroup);
        //写入新的人群包
        super.baseMapper.insertUserGroupToCK(id,sql);
    }
}
